package com.bilibili.lzx.service.config;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson2.JSONArray;
import com.bilibili.lzx.dao.UserFollowingDao;
import com.bilibili.lzx.dao.domain.UserFollowing;
import com.bilibili.lzx.dao.domain.UserMoment;
import com.bilibili.lzx.dao.domain.constant.UserMomentsConstant;
import com.bilibili.lzx.service.UserFollowingService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;

/**
 * @author lzx
 * @date 2024/8/27 21:00
 */

@Configuration
public class RocketMQConfig {
    @Value("${rocketmq.name.server.address}")
    private String nameServerAddr;

    @Resource
    private UserFollowingService userFollowingService;

    @Resource
    private RedisTemplate<String,String> redisTemplate;

    public RocketMQConfig(){
        System.out.println("构造器执行");
    }

    @PostConstruct
    public void post(){
        System.out.println(nameServerAddr);
    }

    @Bean("momentsProducer")
    public DefaultMQProducer momentsProducer() throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer(UserMomentsConstant.MOMENT_GROUP);
        producer.setNamesrvAddr(nameServerAddr);
        producer.start();
        System.out.println("moment-producer启动成功");
        return producer;
    }

    @Bean("momentsConsumer")
    public DefaultMQPushConsumer momentsConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(UserMomentsConstant.MOMENT_GROUP);
        consumer.setNamesrvAddr(nameServerAddr);
        consumer.subscribe(UserMomentsConstant.MOMENT_TOPIC,"*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt messageExt = list.get(0);
                if (messageExt==null){
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                String bodyString = new String(messageExt.getBody());
                UserMoment userMoment = JSONObject.parseObject(bodyString,UserMoment.class);
                Long userId = userMoment.getUserId();
                List<UserFollowing> userFans = userFollowingService.getUserFans(userId);
                for (UserFollowing userFan : userFans) {
                    String key = "user_subscribe_moment_"+userFan.getUserId();
                    String value = redisTemplate.opsForValue().get("key");
                    List<UserMoment> momentList;
                    if(value==null){
                        momentList=new ArrayList<>();
                    }else{
                        momentList = JSONArray.parseArray(value,UserMoment.class);
                    }
                    momentList.add(userMoment);
                    value=JSONArray.toJSONString(momentList);
                    redisTemplate.opsForValue().set(key,value);
                }
                System.out.println("成功消费用户动态消息往redis添加了一条数据"+messageExt.getMsgId());

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //consumer.start();
        System.out.println("moment-consumer启动成功");
        return consumer;
    }
}
